package com.millstein.flume.interceptor;

import com.millstein.util.JsonUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Description
 * @Author tsing
 * @Date 2024-08-12 16:51
 */
public class ETLInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    /**
     * 判断单个事件中的数据是否完整
     * @param event 待判断的事件
     * @return
     */
    @Override
    public Event intercept(Event event) {
        String message = new String(event.getBody(), StandardCharsets.UTF_8);
        if (JsonUtils.isJSONValidate(message)) {
            return event;
        }
        return null;
    }

    /**
     * 批量判断多个事件中的数据是否完整
     * @param list 待判断的事件集合
     * @return
     */
    @Override
    public List<Event> intercept(List<Event> list) {
        return list.stream()
                .filter(item -> intercept(item) != null)
                .collect(Collectors.toList());
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new ETLInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
